package KVService;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.queryparser.classic.ParseException;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
import org.ini4j.Ini;

import TopologyService.Topology;


public class KVHandler implements KVService.Iface{
	
    // Private member variables (obtained from the ini file)
	private ArrayList<Integer> keyBuckets;	// the buckets this node is responsible to store

	// Map that contains the (key,value) pairs
	private Map<Integer, String> kvStore;
	private Topology topology;
	private Lucene lucene;
	private Integer numBuckets;
	
	/**
	 * Constructor for KVHandler object
	 * 
	 * @param KVIni ini section containing kv information
	 * @throws TTransportException 
	 * @throws NumberFormatException 
	 * @throws IOException 
	 */
	public KVHandler(Ini servicesIni, Topology topology) throws NumberFormatException, TTransportException, IOException {
	    
		this.topology = topology;
	    
	    // Get the 'kv' section from the ini file
	    Ini.Section section = servicesIni.get("kv");
	    
	    // Get the key range maximum
        keyBuckets = getKeyBuckets(section.get("keyBuckets"));
        
        // Get the number of buckets in the system
        numBuckets = Integer.parseInt(section.get("numBuckets"));
	    
	    // Initialize the key value store for this handler
	    kvStore = new HashMap<Integer, String>();
	    
	    // Initialize index on values
	    lucene = new Lucene();
	}

	/**
	 * Function to get the buckets that are owned by this process
	 * 
	 * @param bucketString *ini section string representing a comma-separated list of bucket ids
	 * @return
	 */
	private ArrayList<Integer> getKeyBuckets(String bucketString) {
		ArrayList<Integer> bucketList = new ArrayList<Integer>();
		
		String[] tokens = bucketString.split(",");
		
		// Loop over the tokens
        for (String t : tokens) {
            // Remove whitespace
            String entry = t.replaceAll("\\s", "");
            
            // Add the host to the list to be returned
            bucketList.add(Integer.parseInt(entry));
        }
        
        // Return the list of buckets owned
        return bucketList;
	}
	
	/**
	 * Lookup the node responsible for the key, may forward request to neighboring nodes
	 * 
	 * @return id of node responsible for key if key exists in system, else -1
	 * @throws  
	 */
	@Override
	public int lookup(int key) {
	    int retNode = -1;
	    
		// Check if we own the key
	    if (ownsKey(key)) {	        
	    	// key belongs to this node, return my process id 
    		System.out.println("Process " + topology.myID() + " found key " + " in bucket " + Math.abs(key % 4));
    		retNode = topology.myID();
	    }
	    else {
	    	// Send the lookup to our next neighbor, as long as TTL is satisfied?
	    	// TODO: Are we handling TTL here?
		    
		    KVClient kvNext;
		    boolean sent = false;
		    while(!sent){
		    	String ip = topology.getNextIP();
		    	int port = topology.getNextPort();
		    	int routedNode = topology.getNextID();
			    System.out.println("Process " + topology.myID() + " routing lookup for key " + key + " to node " + routedNode);
				try {
					
					kvNext = new KVClient(ip, port);
					retNode = kvNext.lookup(key);
					sent = true;
				} catch (TException e) {
					// fix topology and try again
					topology.failureDetected(routedNode);
				}
		    }
		    
		   
		    System.out.println("Process " + topology.myID() + " found key (" + key + ") at node " + retNode);
	    }
	    
	    return retNode;
	}
	
	public ArrayList<Integer> getKeyBuckets(){
		return this.keyBuckets;
	}
	
	public boolean ownsKey(int key){
		return keyBuckets.contains(Math.abs(key % this.numBuckets));
	}
	
	public void setKeyStore(Map<Integer, String> kvStore){
		this.kvStore = kvStore;
	}
	
	public Map<Integer, String>  getKeyStore(){
		return this.kvStore;
	}

	/**
	 * Get the value corresponding to provided key. Key must be stored at this node.
	 *
	 */
	@Override
	public String get(int key) throws TException {
	    // Return the value for the provided key
		System.out.println("Process " + topology.myID() + " get key: " + key);
	    return kvStore.get(key);
	}
	
	public ArrayList<Integer> searchValues(String query) throws IOException, ParseException{
		
		return lucene.search(query);
	}

	/**
	 * Insert the (key,value) pair in this node. Key must be in this node's key range.
	 */
	@Override
	public synchronized void insert(int key, String value) throws TException {
	    // Store the key,value pair
		System.out.println("Process " + topology.myID() + " insert key: " + key);
		kvStore.put(key, value);
		try {
			lucene.index(key, value);
		} catch (CorruptIndexException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println("Process " + topology.myID() + " storing " + kvStore.size() + " keys");
	}

	/**
	 * Delete the (key,value) pair from this node. Key must be stored at this node.
	 */
	@Override
	public synchronized void deleteKey(int key) throws TException {
	    // Remove the key,value pair
		System.out.println("Process " + topology.myID() + " delete key: " + key);
	    kvStore.remove(key);
	}
}
